package org.lagou.edu.listener;

import com.google.gson.Gson;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.lagou.edu.serialization.dto.OperationInfo;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Service;

import java.util.List;


/**
 * 数据聚合
 * <p>
 * 基于 canal 监听 DB
 * 聚合商品搜素ES
 * 写入商品相关Redis 缓存
 *
 * @author shiding
 */
@Service
@Slf4j
public class CommodityInfoListener {

    /**
     * 基于不同的表 建立不同的监听逻辑
     */
    /**
     * 同步 类目属性
     *
     * @param record
     */
    @KafkaListener(topics = {"canal_sync_category_info"}, containerFactory = "canalKafkaFactory")
    public void canalSyncCategoryInfo(ConsumerRecord<String, List<OperationInfo>> record) {
        record.value().forEach(d -> {
            log.info("canalSyncCategoryInfo listBin table:{}, type:{}, record:{}",
                    d.getType(), d.getType(), new Gson().toJson(d));
            switch (d.getTable()) {
                case "categoryTable":
                    categoryTableSync(d);
                    break;
                case "....其他相关表":
                    // 待实现其他相关表
                    break;
                default:
            }
        });
    }

    /**
     * 同步 商品信息属性
     *
     * @param record
     */
    @KafkaListener(topics = {"canal_sync_commodity_info"}, containerFactory = "canalKafkaFactory")
    public void canalSyncCommodityInfo(ConsumerRecord<String, List<OperationInfo>> record) {
        record.value().forEach(d -> {
            log.info("canalSyncCommodityInfo listBin table:{}, type:{}, record:{}",
                    d.getType(), d.getType(), new Gson().toJson(d));
            switch (d.getTable()) {
                case "commodityTable":
                    categoryTableSync(d);
                    break;
                case "....其他相关表":
                    // 待实现其他相关表
                    break;
                default:
            }
        });
    }

    public void categoryTableSync(OperationInfo operationInfo) {
        // todo 写入相关Es
        esDataEdit(operationInfo);
        // todo 写入相关redis
        redisDataEdit(operationInfo);
    }


    public void esDataEdit(OperationInfo operationInfo) {
        // todo 写入相关redis
    }

    public void redisDataEdit(OperationInfo operationInfo) {
        // todo 写入相关redis
    }

}
